package com.rush.springbootinit.job.cycle;

import cn.hutool.core.collection.CollUtil;
import com.rush.springbootinit.annotation.DistributedLock;
import com.rush.springbootinit.job.EsExitJugment;
import com.rush.springbootinit.mapper.QuestionMapper;
import com.rush.springbootinit.model.dto.question.QuestionEsDTO;
import com.rush.springbootinit.model.entity.Question;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.query.BulkOptions;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

//  取消注释开启任务
@Component
@Slf4j
public class IncSyncQuestionToEs {

    @Resource
    private QuestionMapper questionMapper;

    @Autowired(required = false)
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Resource
    private EsExitJugment esExitJugment;

    /**
     * 每分钟执行一次
     */
    @DistributedLock(key = "testLock", leaseTime = 20000, waitTime = 5000)
    @Scheduled(fixedRate = 60 * 1000)
    public void run() {
        if (!esExitJugment.isElasticsearchConnected()) {
            log.error("Elasticsearch 连接失败, 不进行增量更新");
            return;
        }

        // 查询近 5 分钟内的数据
        long FIVE_MINUTES = 5 * 60 * 1000L;
        Date fiveMinutesAgoDate = new Date(new Date().getTime() - FIVE_MINUTES);
        List<Question> questionList = questionMapper.listQuestionWithDelete(fiveMinutesAgoDate);
        if (CollUtil.isEmpty(questionList)) {
            log.info("没有近五分钟修改的数据，ES 不进行增量同步");
            return;
        }

        List<QuestionEsDTO> questionEsDTOList = questionList.stream()
                .map(QuestionEsDTO::objToDto)
                .collect(Collectors.toList());

        final int pageSize = 500;
        int total = questionEsDTOList.size();
        log.info("IncSyncQuestionToEs start, total: {}", total);

        try {
            for (int i = 0; i < total; i += pageSize) {
                int end = Math.min(i + pageSize, total);
                log.info("Sync from {} to {}", i, end);
                List<IndexQuery> queries = new ArrayList<>();
                for (QuestionEsDTO dto : questionEsDTOList.subList(i, end)) {
                    IndexQuery indexQuery = new IndexQueryBuilder()
                            .withId(dto.getId().toString())
                            .withObject(dto)
                            .build();
                    queries.add(indexQuery);
                }
                elasticsearchRestTemplate.bulkIndex(queries, BulkOptions.defaultOptions().getClass());
            }
        } catch (Exception e) {
            log.error("Error during bulk sync to ES", e);
        }

        log.info("IncSyncQuestionToEs end, total: {}", total);
    }
}
